package com.letv.thrift;

import org.apache.thrift.TProcessorFactory;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TBinaryProtocol.Factory;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadedSelectorServer;
import org.apache.thrift.server.TThreadedSelectorServer.Args;
import org.apache.thrift.transport.TFastFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TNonblockingServerTransport;
import org.slf4j.Logger;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import com.letv.so.service.VideoDetailService;

import serving.GenericDetailServing;
import serving.GenericDetailServing.Iface;
import serving.GenericDetailServing.Processor;

public class ThriftServer implements InitializingBean,DisposableBean {
    private static Logger logger = org.slf4j.LoggerFactory.getLogger(ThriftServer.class);
    private int port;
    private VideoDetailService detailService;
    private SelectorServerThread serverThread;
    private int maxWorkerThreads;
    
    public void setDetailService(VideoDetailService detailService) {
        this.detailService = detailService;
    }
    
    public void setPort(int port) {
        this.port = port;
    }

    public void setMaxWorkerThreads(int maxWorkerThreads) {
        this.maxWorkerThreads = maxWorkerThreads;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
         init();
    }
    protected void init(){
        try {
            GenericDetailServing.Processor<Iface> processor = new GenericDetailServing.Processor<Iface>(detailService);
            serverThread = new SelectorServerThread(processor, port);
            serverThread.start();
            
            GenericDetailServing.Processor<Iface> processor1 = new GenericDetailServing.Processor<Iface>(detailService);
            ServerThread1 serverThread1 = new ServerThread1(processor1, 9099);
            serverThread1.start();
            logger.info("Starting server on port {} successfully.",this.port);
        } catch (Exception e) {
            logger.info("Starting server on port {} error.",this.port,e);
        }
    }
    
    class ServerThread1 extends Thread {
        private TServer server;
        ServerThread1(Processor processor, int port) throws Exception {
          TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(port);
          THsHaServer.Args arg = new THsHaServer.Args(serverTransport);
          arg.protocolFactory(new TCompactProtocol.Factory());
          arg.transportFactory(new TFastFramedTransport.Factory());
          arg.processorFactory(new TProcessorFactory(processor));
          arg.workerThreads(maxWorkerThreads);
          server = new THsHaServer(arg);          
        }

        @Override
        public void run(){
          try{
            server.serve();
          }catch(Exception e){
            //
          }
        }
        
        public void stopServer(){
          server.stop();
        }
      }
    class SelectorServerThread extends Thread {
        private TServer server;
        SelectorServerThread(Processor processor, int port) throws Exception {
          TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(port);
          Factory portFactory = new TBinaryProtocol.Factory(true, true);
          Args args = new TThreadedSelectorServer.Args(serverTransport);
          args.protocolFactory(portFactory);
          args.transportFactory(new TFastFramedTransport.Factory(TFastFramedTransport.DEFAULT_BUF_CAPACITY,
					TFastFramedTransport.DEFAULT_BUF_CAPACITY << 2));
          args.workerThreads(maxWorkerThreads);
          args.maxReadBufferBytes = 1024 * 1024;
          args.processorFactory(new TProcessorFactory(processor));
          server = new TThreadedSelectorServer(args);      
        }

        @Override
        public void run(){
          try{
            server.serve();
          }catch(Exception e){
            //
          }
        }
        
        public void stopServer(){
          server.stop();
        }
      }
    public void close() {
        serverThread.stopServer();
    }

    @Override
    public void destroy() throws Exception {
        close();
    }

}
